package com.example.wxg.stream.custom.source;

import org.apache.flink.streaming.api.functions.source.SourceFunction;

/**
 * @author void
 * @date 2021/11/12 12:04
 * @desc 实现并行度只能为1的自定义DataSource，模拟产生从1开始的数据，每隔1s增加1
 */
public class MyNoParallelSource implements SourceFunction<Long> {
    
    private Long count = 1L;
    private boolean isRunning = true;

    /**
     * 主要方法
     * 启动一个source
     * 大部分情况run方法是个循环，可以源源不断产生数据
     * @param sourceContext
     * @throws Exception
     */
    @Override
    public void run(SourceContext<Long> sourceContext) throws Exception {
        while (isRunning){
            sourceContext.collect(count);
            count++;
            //每秒产生一条数据
            Thread.sleep(1000L);
        }
    }

    /**
     * 停止数据输入
     */
    @Override
    public void cancel() {
        isRunning = false;
    }
}
